package com.hph.flinkcdc;



import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * Flink CDC DataStream方式
 */
public class FlinkCDC_DataStream {
    public static void main(String[] args) throws Exception {

        // 获取flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 开启checkpoint,实现断点续传
        // 设置 3s 的 checkpoint 间隔
        env.enableCheckpointing(3000);
        // 设置 checkpoint 超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 设置 checkpoint 存储位置
        env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");
        // 设置最多可以同时进行多少个 checkpoint
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 设置检查点模式（恰好一次与至少一次）
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // 使用flink cdc构建MySQL Source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("localhost")
                .port(3306)
                .databaseList("test") // 设置捕获的数据库， 如果需要同步整个数据库，请将 tableList 设置为 ".*".
                .tableList("test.t1") // 设置捕获的表
                .username("root")
                .password("123456")
                // 只读取增量数据（不同的flink cdc版本不同默认的选项也不同，一般是StartupOptions.latest()）
                //.startupOptions(StartupOptions.latest())
                // 在首次启动时同步全量数据，后面再次启动从保存点开始同步
                .startupOptions(StartupOptions.initial())
                // 每次都全量
                //.startupOptions(StartupOptions.earliest())
                .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
                .build();


        // 读取数据
        DataStreamSource<String> mySQLSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");

        // 打印
        mySQLSource.print();

        // 启动
        env.execute("Print MySQL Snapshot + Binlog");
    }
}
